package advertising.analyzer

import advertising.AppParams
import advertising.handler.{HBaseHandler, JsonHandler, TagsHandler}
import org.apache.hadoop.hbase.{CellUtil, TableName}
import org.apache.hadoop.hbase.client._
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row

import scala.collection.mutable

//  .--,       .--,
// ( (  \.---./  ) )
//  '.__/o   o\__.'
//     {=  ^  =}
//      >  -  <
//     /       \
//    //       \\
//   //|   .   |\\
//   "'\       /'"_.-~^`'-.
//      \  _  /--'         `
//    ___)( )(___
//   (((__) (__)))    高山仰止, 景行行止; 虽不能至, 心向往之.
//                    High mountains, scenery. Although not to, the heart yearns for it. 
object UserTagsAnalyzer {

  private def mergeDriverId(id1: Array[VertexId], id2: Array[VertexId]) = {
    id1 ++ id2
  }
  
  private def mergeTags(tags1: Array[(String, Int)], tags2: Array[(String, Int)]) = {
    // tagsN: Array[(String, Int)]*
    val map = new mutable.HashMap[String, Int]()
    for (tag <- tags1) {
      val num = map.get(tag._1).getOrElse(0)
      map.put(tag._1, tag._2 + num)
    }
    for (tag <- tags2) {
      val num = map.get(tag._1).getOrElse(0)
      map.put(tag._1, tag._2 + num)
    }
    map.toArray
    // (tags1 ++ tags2).groupBy(_._1).mapValues(_.map(_._2).sum).toArray
  }

  private def getUserTagsVertices(baseAdInfoRDD: RDD[(Array[String], Row)]) = {
    baseAdInfoRDD.flatMap(tuple => {
      val row = tuple._2
      val allTags = TagsHandler.row2AllTags(row)
      // 设备标识
      val driversId = tuple._1
      driversId.map(driverId => {
        // 只让头携带标签
        if (driversId.head.equals(driverId)) {
          (driverId.hashCode.toLong, allTags)
        } else {
          (driverId.hashCode.toLong, Array[(String, Int)]())
        }
      })
    })
  }

  private def getUserEdges(baseAdInfoRDD: RDD[(Array[String], Row)]) = {
    baseAdInfoRDD.flatMap(tuple => {
      val driversId = tuple._1
      val commonDriverId = driversId.head
      driversId.tail.map(driverId => {
        Edge(commonDriverId.hashCode.toLong, driverId.hashCode.toLong, 0)
      })
    })
  }

  private def mergeNewAndOldTags(newTags: Array[(String, Int)], oldTags: Array[(String, Int)]) = {
    val map = new mutable.HashMap[String, Int]()
    for (tag <- oldTags) {
      // 旧标签衰减
      map.put(tag._1, (tag._2 * 0.8).toInt)
    }
    for (tag <- newTags) {
      val num = map.get(tag._1).getOrElse(0)
      map.put(tag._1, tag._2 + num)
    }
    map.toArray
    // (tags1 ++ tags2).groupBy(_._1).mapValues(_.map(_._2).sum).toArray
  }

  private def insert(table: Table, rowKey: Array[Byte], family: Array[Byte], qualifier: Array[Byte], newTags: Array[(String, Int)], result: Result) = {
    val tagsValue = JsonHandler.arrayToJson(newTags)
    val putConf = new Put(rowKey).addColumn(family, qualifier, tagsValue.getBytes())
    table.put(putConf)
  }

  private def update(table: Table, rowKey: Array[Byte], family: Array[Byte], qualifier: Array[Byte], newTags: Array[(String, Int)], result: Result) = {
    val cells = result.rawCells()
    val oldTagsValue = new String(CellUtil.cloneValue(cells(0)))
    val oldTags = JsonHandler.jsonToArray(oldTagsValue)
    // 合并标签
    val mergeTags = mergeNewAndOldTags(newTags, oldTags)
    val tagsValue = JsonHandler.arrayToJson(mergeTags)
    val putConf = new Put(rowKey).addColumn(family, qualifier, tagsValue.getBytes())
    table.put(putConf)
  }

  private def save2HBase(userDriverIdHashCode: VertexId, tags: Array[(String, Int)], connection: Connection) = {
    val table = connection.getTable(TableName.valueOf(AppParams.hbaseTagsTableName))
    val rowKey = String.valueOf(userDriverIdHashCode).getBytes
    val family = AppParams.hbaseTagsTableFamily.getBytes()
    val qualifier = AppParams.hbaseTagsTableFamilyQualifier.getBytes()
    // 构建 get
    val getConf = new Get(rowKey).addColumn(family, qualifier)
    val result = table.get(getConf)
    // 是否已经记录过
    val exists = !result.isEmpty
    if(exists) {
      // 已经记录过，更新记录
      update(table, rowKey, family, qualifier, tags, result)
    } else {
      // 没有记录过，插入记录
      insert(table, rowKey, family, qualifier, tags, result)
    }
    table.close()
    connection.close()
  }

  private def tagsSave2HBase(userTagsCollect: RDD[(VertexId, (Array[VertexId], Array[(String, Int)]))]) = {
    userTagsCollect.map(tuple => {
      (tuple._2._1, tuple._2._2)
    }).foreachPartition(iterator => {
      val connection = HBaseHandler.getConnection
      while (iterator.hasNext) {
        val tuple = iterator.next()
        val userDriverIdHashCodeArr = tuple._1
        val tags = tuple._2
        for(userDriverIdHashCode <- userDriverIdHashCodeArr){
          save2HBase(userDriverIdHashCode, tags, connection)
        }
      }
    })
  }

  private def collectTagsAndSave2HBase(cc: VertexRDD[VertexId], userTagsVertices: RDD[(VertexId, Array[(String, Int)])]) = {
    val userTagsCollect: RDD[(VertexId, (Array[VertexId], Array[(String, Int)]))] = cc.join(userTagsVertices).map(tuple => {
      // graph.connectedComponents 之后
      // (15, 1)
      // (16, 1)
      // (17, 1)
      // join 之后
      // (15, (1, tags))
      // (16, (1, tags))
      // (17, (1, tags))
      // map 之后
      // (1, ((15, 16, 17), tags))
      val commonDriverId = tuple._2._1
      val tags = tuple._2._2
      // 交给公共的 ID
      (commonDriverId, (Array(tuple._1), tags))
    }).reduceByKey((tuple1, tuple2) => {
      // 合并标签
      val driverIds = mergeDriverId(tuple1._1, tuple2._1)
      val tags = mergeTags(tuple1._2, tuple2._2)
      (driverIds, tags)
    })
    tagsSave2HBase(userTagsCollect)
  }

  /**
    * 汇总用户标签，并存储到 HBase 中
    * @param baseAdInfoRDD
    */
  def collectUserTagsAndSave2HBase(baseAdInfoRDD: RDD[(Array[String], Row)]) = {
    val userTagsVertices = getUserTagsVertices(baseAdInfoRDD)
    val userEdges = getUserEdges(baseAdInfoRDD)
    // 图
    val graph = Graph(userTagsVertices, userEdges)
    val cc = graph.connectedComponents().vertices
    collectTagsAndSave2HBase(cc, userTagsVertices)
  }

}
